package realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.google.inject.internal.util.$Join;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.bean.PlayChatperTimesBean;
import realtime.func.AsyncDIMFunction;
import realtime.util.ClickHouseUtil;
import realtime.util.MyKafkaUtil;
import realtime.util.MyWindowFunction;


import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

public class Dws_PlayChapterTimes {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        /*env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC");
        env.setStateBackend(new HashMapStateBackend());*/

        System.setProperty("HADOOP_USER_NAME", "root");

        DataStreamSource<String> kafkaStream = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_video_log", "playchapter"));


        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    out.collect(JSON.parseObject(value));
                }
            }
        });

        SingleOutputStreamOperator<JSONObject> jsonStreamWithWatermark = jsonStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(10)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong("ts");
            }
        }));

        KeyedStream<JSONObject, String> keyedBySidStream = jsonStreamWithWatermark.keyBy(k -> {
           // System.out.println(k.getJSONObject("common").getString("sid"));
            return k.getJSONObject("common").getString("sid");

        });


        SingleOutputStreamOperator<PlayChatperTimesBean> playChatperBeanStream = keyedBySidStream.flatMap(new RichFlatMapFunction<JSONObject, PlayChatperTimesBean>() {

            private ValueState<String> valueState;
            private ValueState<Long> playSecState;

            @Override
            public void open(Configuration parameters) throws Exception {
                StateTtlConfig build = StateTtlConfig.newBuilder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                        .build();

                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("valueState", String.class);
                ValueStateDescriptor<Long> playStateDescriptor = new ValueStateDescriptor<>("playSecState", Long.class);

                valueStateDescriptor.enableTimeToLive(build);
                playStateDescriptor.enableTimeToLive(build);

                valueState = getRuntimeContext().getState(valueStateDescriptor);
                playSecState = getRuntimeContext().getState(playStateDescriptor);

            }

            @Override
            public void flatMap(JSONObject value, Collector<PlayChatperTimesBean> out) throws Exception {
                String mark = valueState.value();
                Long playSecValue = playSecState.value();
                String uid = value.getJSONObject("common").getString("uid");
                Long playSec = value.getJSONObject("appVideo").getLong("play_sec");
                String videoId = value.getJSONObject("appVideo").getString("video_id");
                long times = 0;
                long totalTime = 0;

                Set<String> set = new HashSet<>();
                set.add(uid);

                if (mark == null) {
                    valueState.update("1");
                    times = 1;
                }

                if (playSecValue == null) {
                    playSecState.update(playSec);
                    totalTime = playSec;
                } else {
                    long max = Math.max(playSecValue, playSec);
                    playSecState.update(max);

                    totalTime = max;
                }

                PlayChatperTimesBean build = PlayChatperTimesBean.builder()
                        .videoId(videoId)
                        .userSet(set)
                        .playTimes(times)
                        .playSecSum(totalTime)
                        .build();

                out.collect(build);
            }
        });


        SingleOutputStreamOperator<PlayChatperTimesBean> chapterIdStream = AsyncDataStream.unorderedWait(playChatperBeanStream, new AsyncDIMFunction<PlayChatperTimesBean>("DIM_VIDEO_INFO") {
            @Override
            public void join(PlayChatperTimesBean input, JSONObject dimInfo) {

                String chatId = dimInfo.getString("CHAPTER_ID");
                input.setChapterId(chatId);

            }

            @Override
            public String getKey(PlayChatperTimesBean input) {
                return input.getVideoId();
            }
        }, 1, TimeUnit.MINUTES);

        SingleOutputStreamOperator<PlayChatperTimesBean> chapterNameStream = AsyncDataStream.unorderedWait(chapterIdStream, new AsyncDIMFunction<PlayChatperTimesBean>("DIM_CHAPTER_INFO") {
            @Override
            public void join(PlayChatperTimesBean input, JSONObject dimInfo) {
                input.setChapterName(dimInfo.getString("CHAPTER_NAME"));
            }

            @Override
            public String getKey(PlayChatperTimesBean input) {
                return input.getChapterId();
            }
        }, 1, TimeUnit.MINUTES);

        KeyedStream<PlayChatperTimesBean, String> playChatperBeanStringKeyedStream = chapterNameStream.keyBy(k -> k.getChapterId());


        SingleOutputStreamOperator<PlayChatperTimesBean> reduce = playChatperBeanStringKeyedStream.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<PlayChatperTimesBean>() {
                    @Override
                    public PlayChatperTimesBean reduce(PlayChatperTimesBean value1, PlayChatperTimesBean value2) throws Exception {
                        value1.setPlayTimes(value1.getPlayTimes() + value2.getPlayTimes());
                        value1.getUserSet().addAll(value2.getUserSet());
                        value1.setPlaySecSum(value1.getPlaySecSum() + value2.getPlaySecSum());
                        return value1;
                    }
                }, new WindowFunction<PlayChatperTimesBean, PlayChatperTimesBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<PlayChatperTimesBean> input, Collector<PlayChatperTimesBean> out) throws Exception {
                        PlayChatperTimesBean next = input.iterator().next();
                        MyWindowFunction.getWindowFunction(window, next);
                        next.setPeopleCount((long) next.getUserSet().size());
                        out.collect(next);
                    }
                });


        reduce.print(">>");

        reduce.addSink(ClickHouseUtil.getSinkFunction("insert into dws_play_chapter_times_window values(?,?,?,?,?,?,?,?,?)"));

        env.execute();

    }
}
